21. 多维数据可视化-绘制股票相关性热力图

# 计算简单收益率
# pct_change(): (当前值 - 上期值) / 上期值
simple_returns = df_prices.pct_change().dropna()

# 计算对数收益率
# ln(P_t / P_{t-1}) = ln(P_t) - ln(P_{t-1})
log_returns = np.log(df_prices / df_prices.shift(1)).dropna()

# 输出收益率预览
print('简单收益率(前5行):')
print(simple_returns.head())
print(f'\n对数收益率(前5行):')
print(log_returns.head())

# 收益率统计对比
print(f'\n简单收益率统计:')
print(simple_returns.describe().round(4))
print(f'\n对数收益率统计:')
print(log_returns.describe().round(4))

from scipy.stats import pearsonr  # 在古典假设成立时计算皮尔逊检验
from statsmodels.stats.multitest import multipletests  # 对矩阵内多重检验执行FDR调整

n = len(corr_matrix)
pairs = []
for i in range(n):
    for j in range(i+1, n):
        stock1 = corr_matrix.index[i]
        stock2 = corr_matrix.columns[j]
        corr = corr_matrix.iloc[i, j]
        # pearsonr() 返回 (相关系数, p值)
        _, p_value = pearsonr(
            log_returns[stock1], log_returns[stock2]
        )
        pairs.append({
            '股票1': stock1,
            '股票2': stock2,
            '相关系数': corr,
            '原始P值': p_value
        })

df_pairs = pd.DataFrame(pairs)  # 汇总同一相关矩阵内的全部资产对
df_pairs['FDR_q值'] = multipletests(df_pairs['原始P值'], method='fdr_bh')[1]  # 控制检验族的错误发现率
df_pairs['显著性'] = pd.cut(df_pairs['FDR_q值'], bins=[-1, 0.01, 0.05, 0.10, 1], labels=['***', '**', '*', 'ns'])  # 依据调整后q值生成探索性标记
df_pairs = df_pairs.sort_values('相关系数', ascending=False)  # 按相关强度排列输出
print(df_pairs.to_string(index=False))  # 同时展示效应量、原始p值与FDR结果

import seaborn as sns

plt.figure(figsize=(10, 8))
sns.heatmap(
    corr_matrix,         # 相关系数矩阵
    annot=True,          # 显示数值
    cmap='RdYlGn',       # 红-黄-绿配色
    center=0,            # 颜色中心值为0
    square=True,         # 正方形格子
    linewidths=0.5,      # 格子边框
    fmt='.3f',           # 保留3位小数
    cbar_kws={'label': '皮尔逊相关系数'},
    vmin=-1, vmax=1      # 固定范围[-1,1]
)
plt.title('股票收益率相关性热力图', fontsize=14, pad=20)
plt.xlabel('股票', fontsize=12)
plt.ylabel('股票', fontsize=12)
plt.tight_layout()
plt.show()

from scipy.cluster.hierarchy import linkage

linkage_matrix = linkage(corr_matrix, method='average')
g = sns.clustermap(
    corr_matrix, annot=True, cmap='RdYlGn',
    center=0, square=True, linewidths=0.5,
    fmt='.3f', figsize=(10, 10),
    row_linkage=linkage_matrix,
    col_linkage=linkage_matrix,
    tree_kws={'linewidths': 1.5}
)
g.fig.suptitle('股票相关性聚类分析', fontsize=14, y=0.98)
plt.show()

import networkx as nx

threshold = 0.5  # 只保留|ρ|≥0.5的强相关
G = nx.Graph()
for stock in stocks.values():
    G.add_node(stock)

for i in range(n):
    for j in range(i+1, n):
        corr = corr_matrix.iloc[i, j]
        if abs(corr) >= threshold:
            G.add_edge(
                corr_matrix.index[i],
                corr_matrix.columns[j],
                weight=corr
            )

plt.figure(figsize=(10, 8))
pos = nx.spring_layout(G, k=0.5, seed=42)
degrees = dict(G.degree())

# 节点：大小与连接数成正比
nx.draw_networkx_nodes(
    G, pos,
    node_size=[v * 300 for v in degrees.values()],
    node_color='steelblue', alpha=0.7
)
# 边：绿色=正相关，红色=负相关
weights = [G[u][v]['weight'] for u, v in G.edges()]
edge_colors = ['red' if w < 0 else 'green' for w in weights]
nx.draw_networkx_edges(
    G, pos,
    width=[abs(w) * 3 for w in weights],
    edge_color=edge_colors, alpha=0.6
)
nx.draw_networkx_labels(G, pos, font_size=10)
plt.title(f'股票相关性网络图(阈值={threshold})')
plt.axis('off')
plt.show()

stock_a = '中信证券'
stock_b = '中国平安'
window = 20  # 20日滚动窗口（约1个月）

# 计算滚动相关性
rolling_corr = log_returns[stock_a].rolling(
    window
).corr(log_returns[stock_b])

fig, axes = plt.subplots(2, 1, figsize=(14, 10))

# 子图1：滚动相关系数
axes[0].plot(rolling_corr.index, rolling_corr, linewidth=2)
axes[0].axhline(y=0, color='k', linestyle='--')
axes[0].axhline(y=0.5, color='r', linestyle='--', label='阈值0.5')
axes[0].axhline(y=-0.5, color='r', linestyle='--', label='阈值-0.5')
axes[0].set_title(f'{stock_a}与{stock_b}的滚动相关性')
axes[0].legend()

# 子图2：归一化价格走势
norm_prices = df_prices[[stock_a, stock_b]] / \
    df_prices[[stock_a, stock_b]].iloc[0] * 100
axes[1].plot(norm_prices[stock_a], label=stock_a)
axes[1].plot(norm_prices[stock_b], label=stock_b)
axes[1].set_title('归一化价格走势（初始=100）')
axes[1].legend()
plt.tight_layout()
plt.show()